package com.shujia.flink.tf

import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.streaming.api.scala._

object Demo1Map {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val studentDS: DataStream[String] = env.readTextFile("data/students.txt")

    /**
     * MapFunction: 是一个接口需要实现里面的map方法，在map方法中处理数据
     */
    val mapDS: DataStream[(String, Int)] = studentDS.map(
      new MapFunction[String, (String, Int)] {
        /**
         * map函数每一行数据执行一次，
         *
         * @param value 一行数据
         * @return 返回结果
         */
        override def map(value: String): (String, Int) = {
          val split: Array[String] = value.split(",")
          val clazz: String = split.last
          (clazz, 1)
        }
      }
    )

    mapDS.print()

    env.execute()

  }

}
